package com.roc.rocketmqconsumer.messageType;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

public class DelayMsgConsumer {
    public static void main(String[] args) throws Exception {
        // 1. 创建消费者实例，指定消费者组
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("roc_consumer_group");
        // 2. 设置 NameServer 地址
        consumer.setNamesrvAddr("127.0.0.1:9876");
        // 3. 订阅主题和标签
        consumer.subscribe("TOPIC_DELAY", "*");

        // 4. 注册消息监听器，处理接收到的消息
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                for (MessageExt msg : msgs) {
                    try {
                        String msgBody = new String(msg.getBody(), "UTF-8");
                        System.out.println("接收到延迟消息：" + msgBody);
                        // 5. 处理消息逻辑（如订单关闭、提醒等）
                        // 这里模拟业务处理，实际需根据场景完善
                    } catch (Exception e) {
                        e.printStackTrace();
                        // 6. 消费失败，返回 RECONSUME_LATER 触发重试（根据业务决定是否重试）
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    }
                }
                // 7. 消费成功，返回 CONSUME_SUCCESS
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        // 8. 启动消费者
        consumer.start();
        System.out.println("消费者已启动，等待接收延迟消息……");
    }
}